package com._58city.spark.examples;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.Tuple2;

import com.google.common.base.Splitter;
import com.google.common.collect.Lists;

public class JavaKafkaWordCount
{
  private static Logger log = LoggerFactory.getLogger(JavaKafkaWordCount.class);
  private static final Splitter splitter = Splitter.on(" ");
  
  public static void main(String[] args)
  {
    if (args.length < 4)
    {
      System.err.println("Usage: JavaKafkaWordCount <zkQuorum> <group> <topics> <numThreads>");
      System.exit(1);
    }
    SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount");
    
    JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(2000L));
    
    int numThreads = Integer.parseInt(args[3]);
    Map<String, Integer> topicMap = new HashMap<String, Integer>();
    String[] topics = args[2].split(",");
    for (String topic : topics) {
      topicMap.put(topic, Integer.valueOf(numThreads));
    }
    JavaPairReceiverInputDStream<String, String> messages = KafkaUtils.createStream(jssc, args[0], args[1], topicMap);
    
    JavaPairDStream<String, Integer> wordCounts = messages.flatMapToPair(new PairFlatMapFunction<Tuple2<String,String>, String,Integer>()
    {
		private static final long serialVersionUID = 1L;

		public Iterable<Tuple2<String, Integer>> call(Tuple2<String, String> tuple2)
	      {
			String msg = tuple2._2();
			log.info("map line : "+msg);
//			System.err.println(msg);
			List<Tuple2<String, Integer>> wordsList = Lists.newArrayList();
			for(String lineString : splitter.split(msg)){
				log.info("map word: "+lineString);
				wordsList.add(new Tuple2<String, Integer>(lineString, Integer.valueOf(1)));
			}
	        return wordsList;
	      }
    }).reduceByKey(new Function2<Integer, Integer, Integer>()
    {
		private static final long serialVersionUID = 1L;
		public Integer call(Integer i1, Integer i2)
	      {
	        return Integer.valueOf(i1.intValue() + i2.intValue());
	      }
    });
    wordCounts.print();
    jssc.start();
    jssc.awaitTermination();
  }
}
